﻿using System;
using System.Threading.Tasks;
using Streamiz.Kafka.Net;
using Streamiz.Kafka.Net.SerDes;
using Streamiz.Kafka.Net.Stream;
using Streamiz.Kafka.Net.Table;

namespace BASIC_Streamiz.Kafka
{
    class Program
    {
        static async Task Main(string[] args)
        {
            // Stream configuration
            var config = new StreamConfig<StringSerDes, StringSerDes>();
            config.ApplicationId = "test-topic";
            config.BootstrapServers = "192.168.2.150:9092";

            StreamBuilder builder = new StreamBuilder();

            // Stream "test" topic with filterNot condition and persist in "test-output" topic.
            builder.Stream<string, string>("test")
                .FilterNot((k, v) => v.Contains("test"))
                .To("test-topic");

            // Create a table with "test-ktable" topic, and materialize this with in memory store named "test-store"
            builder.Table("test-ktable", InMemory<string,string>.As("test-store"));

            // Build topology
            Topology t = builder.Build();

            // Create a stream instance with toology and configuration
            KafkaStream stream = new KafkaStream(t, config);

            // Subscribe CTRL + C to quit stream application
            Console.CancelKeyPress += (o, e) =>
            {
                stream.Dispose();
            };

            // Start stream instance with cancellable token
            await stream.StartAsync();
        }
    }
}
